package org.apache.ocean.name;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

import net.sourceforge.jsorter.ReflectColumns;
import net.sourceforge.jsorter.SortReflect;
import net.sourceforge.jsorter.Sorter;
import net.sourceforge.jsorter.SorterConstants;

import org.apache.commons.lang.StringUtils;
import org.apache.ocean.ClusterConstants;
import org.apache.ocean.ClusterUtil;

public class NodeUtil {
	public static long MAX_REPORT_EXPIRED_DURATION = 1000 * 10;
	public static long DURATION_AS_MASTER_FOR_AGREEMENT = 1000*5;
  
	public static boolean verifyNode(Node node, Date nameServerTimestamp) {
		if (!StringUtils.isBlank(node.id)) {
			return false;
		}
		if (!NodeUtil.verifyCellIDs(node)) {
			return false;
		}
		if (NodeUtil.isReportDateExpired(node, nameServerTimestamp)) {
			return false;
		}
		return true;
	}
	
	public static boolean verifyCellIDs(Node node) {
		if (!StringUtils.equals(node.assignedCellID, node.reportedCellID)) {
			return false;
		}
		return true;
	}
	
	public static Node obtainAgreedMasterNode(List<Node> incoming, Date serverTimestamp, Logger log) throws Exception {
		List<Node> nodes = filterBadNodes(incoming, serverTimestamp);
		int uniqueMasterCount = getUniqueReportedMasterCount(nodes);
		if (uniqueMasterCount != 1) {
			if (log != null) log.severe("uniqueMasterCount: "+uniqueMasterCount);
			return null;
		}
		try {
			String sameMasterID = verifySlavesHaveSameMaster(nodes, DURATION_AS_MASTER_FOR_AGREEMENT, serverTimestamp, log);
			Node masterNode = (Node) ClusterUtil.reflectFindByField("id", sameMasterID, nodes);
			return masterNode;
		} catch (Exception exception) {
			if (log != null) log.log(Level.SEVERE, exception.getMessage()+" nodes don't have same master: " + nodes);
		}
		return null;
	}
	
	public static String verifySlavesHaveSameMaster(List<Node> nodes, long desiredDurationAsMaster, Date serverTimestamp, Logger log) throws Exception {
		List nonMasterStatusList = ClusterUtil.reflectFindFields("reportedStatus", ClusterConstants.SLAVE, nodes);
		nonMasterStatusList.addAll(ClusterUtil.reflectFindFields("reportedStatus", ClusterConstants.NOTHING, nodes));
		List masterStatusList = ClusterUtil.reflectFindFields("reportedStatus", ClusterConstants.MASTER, nodes);
    if (nonMasterStatusList.size() == 0 && 	masterStatusList.size() == 1) {
    	Node masterNode = (Node)ClusterUtil.reflectFindByField("reportedStatus", ClusterConstants.MASTER, nodes);
    	if (log != null) log.info("no slaves, one master");
    	return masterNode.id;
    }
		String masterID = null;
		// make sure all nodes have chosen the same master id
		for (Node node : nodes) {
			if (StringUtils.equals(ClusterConstants.SLAVE, node.reportedStatus) ||
					StringUtils.equals(ClusterConstants.NOTHING, node.reportedStatus)) {
				if (masterID != null && !StringUtils.equals(node.reportedMasterID, masterID)) {
					throw new Exception("masters do not match node id: " + node.id);
				}
				if (node.reportedMasterIDSinceDate == null) {
					continue;
				}
					//throw new Exception("node " + node.id + " reportedMasterIDSinceDate null");
				// master for at least desiredDurationAsMaster
				if (node.reportedMasterIDSinceDate != null) {
					long durationAsMaster = serverTimestamp.getTime() - node.reportedMasterIDSinceDate.getTime();
					if (durationAsMaster > desiredDurationAsMaster) {
						masterID = node.reportedMasterID;
					}
				} else {
				}
			}
		}
		return masterID;
	}
	
	public static boolean isReportDateExpired(Node node, Date serverTimestamp) {
		if (node.reportDate == null) return true;
		long duration = serverTimestamp.getTime() - node.reportDate.getTime();
		return (duration > MAX_REPORT_EXPIRED_DURATION);
	}
	
	public static int getUniqueReportedMasterCount(List<Node> nodes) {
		Set<String> set = new HashSet<String>();
		for (Node node : nodes) {
			if (node.reportedMasterID != null) set.add(node.reportedMasterID);
		}
		return set.size();
	}
	
	public static Node chooseMaster(List<Node> incoming, Date serverTimestamp) throws Exception {
		List<Node> nodes = filterExpiredNodes(incoming, serverTimestamp);
		nodes = filterOfflineNodes(nodes);
		ReflectColumns reflectColumns = new ReflectColumns();
		reflectColumns.add("highestCellTransactionIDDatabase", SortReflect.field("highestCellTransactionIDDatabase"), SorterConstants.DESCENDING_ORDER);
		reflectColumns.add("dateCreated", SortReflect.field("dateCreated"), SorterConstants.ASCENDING_ORDER);
		Sorter sorter = Sorter.createReflect(reflectColumns, nodes, SorterConstants.NULLS_ARE_LEAST);
		List<Node> sortedNodes = sorter.sortReflect();
		for (Node node : sortedNodes) {
			return node;
		}
		return null;
	}

	public static boolean isOnline(Node incomingNode) {
		if (StringUtils.equals(ClusterConstants.MASTER, incomingNode.reportedStatus) 
				|| StringUtils.equals(ClusterConstants.SLAVE, incomingNode.reportedStatus)
				|| StringUtils.equals(ClusterConstants.NOTHING, incomingNode.reportedStatus)) {
			return true;
		} else
			return false;
	}
  
	public static List<Node> filterBadNodes(List<Node> incoming, Date serverTimestamp) {
		List<Node> offlineFiltered = filterOfflineNodes(incoming);
		return filterExpiredNodes(offlineFiltered, serverTimestamp);
	}
	
	public static List<Node> filterOfflineNodes(List<Node> incoming) {
		List<Node> outgoing = new ArrayList<Node>();
		for (Node incomingNode : incoming) {
			if (isOnline(incomingNode)) {
				outgoing.add(incomingNode);
			}
		}
		return outgoing;
	}

	public static List<Node> filterExpiredNodes(List<Node> incoming, Date serverTimestamp) {
		List<Node> outgoing = new ArrayList<Node>();
		for (Node incomingNode : incoming) {
			if (incomingNode.reportDate != null) {
				long duration = serverTimestamp.getTime() - incomingNode.reportDate.getTime();
				if (duration < MAX_REPORT_EXPIRED_DURATION) {
					outgoing.add(incomingNode);
				}
			}
		}
		return outgoing;
	}
}
